//package cn.com.zetatech.loader.resolver.kafka.consumer;
//
//import cn.com.zetatech.loader.entity.ResolverHuangshanMessage;
//import cn.com.zetatech.loader.resolver.service.DefectResolverService;
//import com.google.gson.Gson;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;
//
///**
// * defect解析逻辑
// */
//@Slf4j
//@Component
//public class DefectResolveTaskConsumer {
//    @Autowired
//    private Gson gson;
//    @Autowired
//    private DefectResolverService defectResolverService;
//
//    @KafkaListener(topics = "#{'${resolver.topic}'}",
//            concurrency = "${resolver.consumer.concurrency:2}",
//            groupId = "${resolver.consumer.groupId:defect_consume6}")
//    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
//        try {
//            String value = record.value();
//            String topic = record.topic();
//            String key = record.key();
//
//            System.out.println(topic);
//            System.out.println(key);
//            if(!StringUtils.isEmpty(value)) {
//                // 对消息进行解码
//                log.info("<============================ 消费Kafka defect消息  ============================>");
//                ResolverHuangshanMessage message = gson.fromJson(value, ResolverHuangshanMessage.class);
//                if(message != null) {
//                    log.info("<============================ 开始转移任务 ============================>");
//                    defectResolverService.execute(message);
//                }
//            }
//        } catch (Exception e) {
//            log.error(e.getMessage(), e);
//        } finally {
//            ack.acknowledge();
//        }
//    }
//}
